#include <sys/time.h>
#include <stdlib.h>
#include <pthread.h>
#include <sys/types.h>
#include "blockingqueue.h"

int bq_push(blockingqueue this,void *data);

void* bq_pop(blockingqueue this,u_int64_t timeout);

void* bq_top(blockingqueue this,u_int64_t timeout);

void* bq_try_pop(blockingqueue this);

void* bq_try_top(blockingqueue this);

int bq_is_empty(blockingqueue this);

void bq_clear(blockingqueue this);

void destory_blockingqueue(blockingqueue this);

blockingqueue create_blockingqueue(){
    blockingqueue this=(blockingqueue)calloc(1,sizeof(blockingqueue_t));
    this->q=create_queue();
    this->mutex=(pthread_mutex_t)PTHREAD_MUTEX_INITIALIZER;
    this->cond=(pthread_cond_t)PTHREAD_COND_INITIALIZER;

    this->is_empty=bq_is_empty;
    this->push=bq_push;
    this->pop=bq_pop;
    this->top=bq_top;
    this->try_pop=bq_try_pop;
    this->try_top=bq_try_top;
    this->clear=bq_clear;
    this->destory=destory_blockingqueue;
    return this;
} 

//non-blocking 
int bq_push(blockingqueue this,void *data){
    pthread_mutex_lock(&this->mutex);
    int ret = (this->q->push(this->q,data));
    pthread_mutex_unlock(&this->mutex);
    pthread_cond_broadcast(&this->cond);
    return ret;
}

//blocking
void* bq_pop(blockingqueue this,u_int64_t timeout){
    pthread_mutex_lock(&this->mutex);
    if(bq_is_empty(this)){
        struct timeval now;
        gettimeofday(&now,NULL);
        u_int64_t ns=now.tv_usec*(int)1e3+timeout;

        struct timespec outtime;
        outtime.tv_sec = now.tv_sec + (ns / (int)1e9);
        outtime.tv_nsec = ns % (int)1e9;

        pthread_cond_timedwait(&this->cond,&this->mutex,&outtime);
    }
    void* data=this->q->pop(this->q);
    pthread_mutex_unlock(&this->mutex);
    return data;
}

//阻塞，等待时间单位为纳秒
void* bq_top(blockingqueue this,u_int64_t timeout){
    pthread_mutex_lock(&this->mutex);
    if(bq_is_empty(this)){
        struct timeval now;
        gettimeofday(&now,NULL);
        u_int64_t ns=now.tv_usec*(int)1e3+timeout;

        struct timespec outtime;
        outtime.tv_sec = now.tv_sec + (ns / (int)1e9);
        outtime.tv_nsec = ns % (int)1e9;

        pthread_cond_timedwait(&this->cond,&this->mutex,&outtime);
    }
    void* data=this->q->top(this->q);
    pthread_mutex_unlock(&this->mutex);
    return data;
}

//non-blocking
void* bq_try_pop(blockingqueue this){
    pthread_mutex_lock(&this->mutex);
    void *data=this->q->pop(this->q);
    pthread_mutex_unlock(&this->mutex);
    return data;
}

//non-blocking
void* bq_try_top(blockingqueue this){
    pthread_mutex_lock(&this->mutex);
    void* data=this->q->top(this->q);
    pthread_mutex_unlock(&this->mutex);
    return data;
}

int bq_is_empty(blockingqueue this){
    return this->q->is_empty(this->q);
}

void bq_clear(blockingqueue this){
    pthread_mutex_lock(&this->mutex);
    this->q->clear(this->q);
    pthread_mutex_unlock(&this->mutex);
    return ;
}

void destory_blockingqueue(blockingqueue this){
    if(this==NULL)return ;
    this->q->destroy(this->q);
    pthread_mutex_destroy(&this->mutex);
    pthread_cond_destroy(&this->cond);
    free(this);
}